package com.dx_society.service.impl;

import cn.hutool.core.lang.Snowflake;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.dx_society.domain.Message;
import com.dx_society.domain.User;
import com.dx_society.dto.MsgDTO;
import com.dx_society.dto.Result;
import com.dx_society.mapper.UserMapper;
import com.dx_society.service.MessageService;
import com.dx_society.mapper.MessageMapper;
import com.dx_society.utils.UserHolder;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * @author 韶光善良君
 * @description 针对表【dx_message(消息表)】的数据库操作Service实现
 * @createDate 2024-07-01 10:15:16
 */
@Slf4j
@Service
public class MessageServiceImpl extends ServiceImpl<MessageMapper, Message>
        implements MessageService {
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private UserMapper userMapper;
    @Resource
    private Snowflake snowflake;
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("mqadd.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    @Override
    public Result sendMessage(Long receiveUserId, String message) {
        Long userId;
        try {
            userId = UserHolder.getUser().getUserid();
        } catch (Exception e) {
            return Result.fail("用户未登录");
        }
        Long msgId = snowflake.nextId();
        Message msg = new Message();
        msg.setId(msgId);
        msg.setReceiveUserId(receiveUserId);
        msg.setMessage(message);
        Date now = Date.from(LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant());
        msg.setSendtime(now);
        msg.setSendUserId(userId);
        save(msg);

        // 保存到redis对应收消息者的收件箱中，以及发送者的发件箱中 lua脚本
        stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                receiveUserId.toString(),
                userId.toString(),
                message,
                now.toString(),
                msgId.toString()
        );
        log.info("MQ添加成功");
        return Result.ok("发送成功");
    }

    @Override
    public Result queryUnreadMessage() {
        Long userId;
        try {
            userId = UserHolder.getUser().getUserid();
        } catch (Exception e) {
            return Result.fail("用户未登录");
        }
        String key = "stream.inbox." + userId;
        if (Boolean.FALSE.equals(stringRedisTemplate.hasKey(key))) {
            return Result.fail("您还没有收到过消息");
        }
        List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                Consumer.from("g1", "c1"),
                StreamOffset.create(key, ReadOffset.lastConsumed())
        );
//        这里我直接使用了read，相当于pending完所有未读消息后直接全部ack了，应该改为pending。
//        PendingMessagesSummary g1 = stringRedisTemplate.opsForStream().pending(key, "g1");
//        不知道stringRedisTemplate是不是这样写的，可能要加一下范围，实在不行可以使用Lua脚本
        if (list == null || list.isEmpty()) {
            // list为空，说明没有未读消息
            return Result.fail("你没有未读的消息");
        }
        List<MsgDTO> msgDTOList = new ArrayList<>();
        for (MapRecord<String, Object, Object> record : list) {
            Map<Object, Object> value = record.getValue();
            // 解析消息内容
            String sender = (String) value.get("sender");
            String messageContent = (String) value.get("message");
            String sendTimeStr = (String) value.get("sendtime");
            Long senderId = Long.parseLong(sender);

            User user = userMapper.selectById(senderId);
            String senderName = user.getUsername();
            // 创建 DateTimeFormatter
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("EEE MMM dd HH:mm:ss z yyyy", Locale.ENGLISH);
            ZoneId zoneId = ZoneId.of("Asia/Shanghai");
            Date sendTime = Date.from(LocalDateTime.parse(sendTimeStr, formatter)
                    .atZone(zoneId)
                    .toInstant());
            MsgDTO msgDTO = new MsgDTO();
            msgDTO.setSendUserId(senderId);
            msgDTO.setSendUserName(senderName);
            msgDTO.setContext(messageContent);
            msgDTO.setSendtime(sendTime);
            msgDTOList.add(msgDTO);

            stringRedisTemplate.opsForStream().acknowledge(key, "g1", record.getId().getValue());
//            stringRedisTemplate.opsForStream().delete(key, record.getId());
        }
        // 将msgDTOList按照日期降序排列
        msgDTOList.sort(Comparator.comparing(MsgDTO::getSendtime).reversed());

        return Result.ok(msgDTOList);
    }

    @Override
    public Result queryOutMsg() {
        Long userId;
        try {
            userId = UserHolder.getUser().getUserid();
        } catch (Exception e) {
            return Result.fail("用户未登录");
        }
        String key = "list.outbox." + userId;
        if (Boolean.FALSE.equals(stringRedisTemplate.hasKey(key))) {
            return Result.fail("你还没有发送过消息");
        }
        List<String> msgStrList = stringRedisTemplate.opsForList().range(key, 0, -1);
        if(msgStrList!=null&&!msgStrList.isEmpty()){
            List<MsgDTO> msgDTOList = new ArrayList<>();
            for (String msgStr : msgStrList) {
                String[] msgArr = msgStr.split("\\|");
                Long msgId = Long.parseLong(msgArr[0]);
                Long receiverId = Long.parseLong(msgArr[1]);
                String message = msgArr[2];
                // 创建 DateTimeFormatter
                // 创建 DateTimeFormatter
                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("EEE MMM dd HH:mm:ss z yyyy", Locale.ENGLISH);
                ZoneId zoneId = ZoneId.of("Asia/Shanghai");
                Date sendTime = Date.from(LocalDateTime.parse(msgArr[3], formatter)
                        .atZone(zoneId)
                        .toInstant());
                MsgDTO msgDTO = new MsgDTO();
                msgDTO.setSendUserId(receiverId);
                msgDTO.setSendUserName(userMapper.selectById(receiverId).getUsername());
                msgDTO.setContext(message);
                msgDTO.setSendtime(sendTime);

                msgDTOList.add(msgDTO);

            }
            // 将msgDTOList按照日期降序排列
            msgDTOList.sort(Comparator.comparing(MsgDTO::getSendtime).reversed());
            return Result.ok(msgDTOList);
        }
        return Result.fail("你没有发送过消息");
    }
    // 获取收件箱
    @Override
    public Result queryInbox() {
        Long userId;
        try {
            userId = UserHolder.getUser().getUserid();
        } catch (Exception e) {
            return Result.fail("用户未登录");
        }
        List<Message> receiveUserId = query().eq("receive_user_id", userId).list();
        List<MsgDTO> msgDTOList = new ArrayList<>();
        for (Message message : receiveUserId) {
            MsgDTO msgDTO = new MsgDTO();
            Long sendUserId = message.getSendUserId();
            User user = userMapper.selectById(sendUserId);
            String username = user.getUsername();
            msgDTO.setSendUserId(sendUserId);
            msgDTO.setSendUserName(username);
            msgDTO.setContext(message.getMessage());
            msgDTO.setSendtime(message.getSendtime());
            msgDTOList.add(msgDTO);
        }
        // 将msgDTOList按照日期降序排列
        msgDTOList.sort(Comparator.comparing(MsgDTO::getSendtime).reversed());
        return Result.ok(msgDTOList);
    }

    @Override
    public Result queryRecent() {
        Long userId;
        try {
            userId = UserHolder.getUser().getUserid();
        } catch (Exception e) {
            return Result.fail("用户未登录");
        }
        String key = "stream.inbox." + userId;
        if (Boolean.FALSE.equals(stringRedisTemplate.hasKey(key))) {
            return Result.fail("您还没有收到过消息");
        }
        List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().range(key, Range.unbounded());
        if (list == null || list.isEmpty()) {
            // list为空，说明没有消息
            return Result.fail("您还没有收到过消息");
        }
        List<MsgDTO> msgDTOList = new ArrayList<>();
        for (MapRecord<String, Object, Object> record : list) {
            Map<Object, Object> value = record.getValue();
            // 解析消息内容
            String sender = (String) value.get("sender");
            String messageContent = (String) value.get("message");
            String sendTimeStr = (String) value.get("sendtime");
            Long senderId = Long.parseLong(sender);

            User user = userMapper.selectById(senderId);
            String senderName = user.getUsername();
            // 创建 DateTimeFormatter
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("EEE MMM dd HH:mm:ss z yyyy", Locale.ENGLISH);
            ZoneId zoneId = ZoneId.of("Asia/Shanghai");
            Date sendTime = Date.from(LocalDateTime.parse(sendTimeStr, formatter)
                    .atZone(zoneId)
                    .toInstant());
            MsgDTO msgDTO = new MsgDTO();
            msgDTO.setSendUserId(senderId);
            msgDTO.setSendUserName(senderName);
            msgDTO.setContext(messageContent);
            msgDTO.setSendtime(sendTime);
            msgDTOList.add(msgDTO);
        }
        // 将msgDTOList按照日期降序排列
        msgDTOList.sort(Comparator.comparing(MsgDTO::getSendtime).reversed());
        return Result.ok(msgDTOList);
    }

}




